package com.shujia.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

object Demo1Producer {
  def main(args: Array[String]): Unit = {
    /**
      *
      * 1、创建生产者，和kafka的集群建立网络链接
      */

    val properties = new Properties
    //指定kafka链接参数

    //指定kafka briker列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")

    //指定key value序列化类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    val kafkaProducer = new KafkaProducer[String, String](properties)


    /**
      * 将数据发生到kafka集群中
      * 分区策略默认是轮询
      * 时间戳默认是当前时间
      * key默认是null
      *
      */
    val record = new ProducerRecord[String, String]("test_topic2", "java")

    kafkaProducer.send(record)

    kafkaProducer.flush()

    //关闭链接
    kafkaProducer.close()

  }

}
